package com.boyi.shared.kafka;

import com.boyi.shared.util.SerializeUtil;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

public class KafkaObjectDeseralizer implements Deserializer<Object> {

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
    }

    @Override
    public Object deserialize(String topic, byte[] bytes) {
        if (bytes == null) {
            return null;
        }
        Object obj = null;
        try {
            if (bytes != null) {
                obj = SerializeUtil.deserialize(bytes);
            }
            return obj;
        } catch (IOException e) {
            throw new SerializationException("Can't deserialize data [" + Arrays.toString(bytes) +
                    "] from topic [" + topic + "]", e);
        } catch (ClassNotFoundException e) {
            throw new SerializationException("Can't deserialize data [" + Arrays.toString(bytes) +
                    "] from topic [" + topic + "]" + " due to " + obj.getClass() + " not found", e);
        }
    }

    @Override
    public void close() {

    }
}

